package org.tradesystem.tickernews;

import java.net.URL;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

import org.activequant.container.services.ITimeStampService;
import org.activequant.core.domainmodel.InstrumentSpecification;
import org.activequant.core.domainmodel.data.TickerNews;
import org.activequant.core.types.TimeFrame;
import org.activequant.core.types.TimeStamp;
import org.activequant.data.annotations.Vendor;
import org.activequant.data.retrieval.ISubscription;
import org.activequant.data.retrieval.ITickerNewsSubscriptionSource;
import org.activequant.data.retrieval.SubscriptionSourceBase;
import org.activequant.util.pattern.events.Event;
import org.activequant.util.pattern.events.IEventListener;
import org.activequant.util.tools.UniqueDateGenerator;
import org.apache.log4j.Logger;
import org.encog.bot.rss.RSS;
import org.encog.bot.rss.RSSItem;


/**
* 
* RSS tickerNews source.
* <br>
* RSSTickerNewsSource extends SubscriptionSourceBase&lt;TickerNews&gt; implements ITickerNewsSubscriptionSource.<br>
* Holds the following associated variables:
* <ul>
* <li>connection(IBTwsConnection)</li>
* </ul>
* Holds the following inherited variables:
* <ul>
* <li>activeBackends(Map&lt;Topic, Subscription&gt;)</li>
* <li>activeSubscriptions(Queue&lt;SubscriptionDelegate&gt;)</li>
* </ul>
*/
@Vendor("Yahoo")
public class RSSTickerNewsSource extends SubscriptionSourceBase<TickerNews> implements ITickerNewsSubscriptionSource {
	/**
	 * private final ITimeStampService timeStampService;
	 */
	private final ITimeStampService timeStampService;
	
	private final ExecutorService executor;

	protected final static Logger log = Logger.getLogger(RSSTickerNewsSource.class);
	/**
	 * constructs an RSSTickerNewsSource(extends SubscriptionSourceBase&lt;TickerNews&gt; implements ITickerNewsSubscriptionSource)
	 * @param monitor
	 */
	public RSSTickerNewsSource(ITimeStampService timeStampService) {
		super();
		this.timeStampService=timeStampService;
		this.executor = Executors.newCachedThreadPool();
	}
	/**
	 * Returns/creates a ISubscription&lt;TickerNews&gt; for the given spec(InstrumentSpecification) and TIMEFRAME_1_TICK.
	 */
	public ISubscription<TickerNews> subscribe(InstrumentSpecification spec) {
		return subscribe(spec, TimeFrame.TIMEFRAME_1_TICK);
	}
	/**
	 * TickerNewsSubscription extends Subscription. Holds the following associated variables:
	 * <ul>
	 * <li>spec(InstrumentSpecification)</li>
	 * <li>listener(IEventListener&lt;TickerNews&gt;)</li>
	 * </ul>
	 * Holds the following inherited variables:
	 * <ul>
	 * <li>isActive(AtomicBoolean)</li>
	 * <li>refCount(AtomicInteger)</li>
	 * <li>instrument(InstrumentSpecification)</li>
	 * <li>event(Event&lt;TickerNews&gt;)</li>
	 * </ul>
	 * @author Dimitar
	 *
	 */
	private class TickerNewsSubscription extends Subscription {
		/**
		 * private final InstrumentSpecification spec;
		 */
		private final InstrumentSpecification spec;
		/**
		 * constructs a TickerNewsSubscription(extends Subscription) using the given s(InstrumentSpecification) to set its associated spec(InstrumentSpecification)
		 * @param s
		 */
		TickerNewsSubscription(InstrumentSpecification s) {
			spec = s;
		}
		/**
		 * private final IEventListener&lt;TickerNews&gt; listener = new IEventListener&lt;TickerNews&gt;() {...}<br/>
		 * A listener that re-fires any event(TickerNews)
		 */
		private final IEventListener<TickerNews> listener = new IEventListener<TickerNews>() {
			public void eventFired(TickerNews event) {
				fireEvent(event);
			}
		};
		/**
		 * Subscribes the associated listener(IEventListener&lt;TickerNews&gt;) for real time ticker news from the rss feed for the associated spec(InstrumentSpecification)
		 */
		@Override
		protected void handleActivate() {
			log.info("subscribe to TickerNews stream: " + spec.getSymbol().toString()+" "+spec.getSecurityType().toString()+"@"+spec.getExchange());
			synchronized(tickerNewsWrappers) {
				TickerNewsWrapper w = tickerNewsWrappers.get(spec);
				if(w == null) {
					w = new TickerNewsWrapper(spec, timeStampService.currentTimeStamp());
					tickerNewsWrappers.put(spec, w);
					executor.execute(w);
				}

				w.tickerNewsEvent.addEventListener(listener);
			}
			
		}
		/**
		 * Unsubscribes the associated listener(IEventListener&lt;TickerNews&gt;) from listening to real time ticker news from the RSS feed for the associated spec(InstrumentSpecification)
		 */
		@Override
		protected void handleCancel() {
			log.info("unsubscribe from TickerNews stream: " + spec.getSymbol().toString()+" "+spec.getSecurityType().toString()+"@"+spec.getExchange());

			synchronized(tickerNewsWrappers) {
				TickerNewsWrapper w = tickerNewsWrappers.get(spec);
				if(w == null) {
					log.warn("not subscribed");
					return;
				}

				w.tickerNewsEvent.removeEventListener(listener);
				if(w.tickerNewsEvent.isEmpty()) {
					w.needsToStop.set(true);
					tickerNewsWrappers.remove(spec);
				}
			}
		}
	}
	/**
	 * creates and returns a TickerNewsSubscription(extends Subscription) using the given spec(InstrumentSpecification) to set its associated spec(InstrumentSpecification).<br/>
	 * The given timeFrame(TimeFrame) is not used.
	 */
	@Override
	protected TickerNewsSubscription createSubscription(InstrumentSpecification spec, TimeFrame timeFrame) {
		return new TickerNewsSubscription(spec);
	}
	
	private static class TickerNewsWrapper implements Runnable{		
		/**
		 * private final InstrumentSpecification spec;<br/>
		 * request specification
		 */
		private final InstrumentSpecification spec;		
		/**
		 * private TimeStamp activatonTime;
		 */
		private TimeStamp activatonTime;
		/**
		 * private RSS rss;
		 */
		private RSS rss;
		/**
		 * public final AtomicBoolean needsToStop = new AtomicBoolean(false);
		 */
		public final AtomicBoolean needsToStop = new AtomicBoolean(false);
		/**
		 * private final TickerNews tickerNews;<br/>
		 * accumulate tickerNews parameters here
		 */
		private final TickerNews tickerNews;
		/**
		 * private final UniqueDateGenerator tickerNewsGenerator = new UniqueDateGenerator();
		 */
		private final UniqueDateGenerator tickerNewsGenerator = new UniqueDateGenerator(); 
		/**
		 * private String lastTitle;<br/>
		 * I need this because for some reason it sometimes keeps reading the same news with different timestamps?
		 */
		private String lastTitle;
		/**
		 * constructs a TickerNews using the given spec(InstrumentSpecification) to set its associated spec(InstrumentSpecification)
		 * @param spec
		 */
		TickerNewsWrapper(InstrumentSpecification spec, TimeStamp activatonTime) {
			this.spec = spec;
//			this.rss=new RSS();
			this.activatonTime=activatonTime;
			this.tickerNews = new TickerNews(spec);
		}
		/**
		 * public final Event&lt;TickerNews&gt; tickerNewsEvent = new Event&lt;TickerNews&gt;();<br/>
		 * tickerNews event dispatcher
		 */
		public final Event<TickerNews> tickerNewsEvent = new Event<TickerNews>();
		
		private void distributeTickerNews(TickerNews tickerNews) {
			try {
				TickerNews q = tickerNews.clone();
				q.setTimeStamp(tickerNewsGenerator.generate(new Date()));
				tickerNewsEvent.fire(q);
				lastTitle=q.getTitle();
			} catch (Exception e) {
				e.printStackTrace();
				log.error(e);
			}
		}
		
		public void run() {
//			Logging.stopConsoleLogging();
			while (!needsToStop.get()) {				
				String symbol=spec.getSymbol().toString();
				try {
					URL url = new URL("http://finance.yahoo.com/rss/headline?s="+symbol.toLowerCase()+"&rand="+Math.random());
					this.rss=new RSS();
				    rss.load(url, 15000);
					for (RSSItem item : rss.getItems()) {
						TimeStamp timeStamp = new TimeStamp(item.getDate());
						if (timeStamp.isBefore(activatonTime)||timeStamp.isEqual(activatonTime)||(item.getTitle()!=null&&item.getTitle().equals(lastTitle))) {
							break;
						}
						this.tickerNews.setTitle(item.getTitle());
						this.tickerNews.setTimeStamp(timeStamp);
						this.tickerNews.setLink(item.getLink());
						this.tickerNews.setDescription(item.getDescription());
						if(activatonTime.isBefore(timeStamp)){
							activatonTime=timeStamp;
						}
						distributeTickerNews(tickerNews);
					}
					Thread.sleep(15000);

				} catch (Exception x) {
					log.warn("Exception while checking news", x);
				}
			}
		}
	}
	
	/**
	 * private final Map&lt;InstrumentSpecification,TickerNewsWrapper&gt; tickerNewsWrappers = new HashMap&lt;InstrumentSpecification,TickerNewsWrapper&gt;();
	 */
	private final Map<InstrumentSpecification,TickerNewsWrapper> tickerNewsWrappers = new HashMap<InstrumentSpecification,TickerNewsWrapper>();
}